package com.zeny.express.framework.mq;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanNameAware;

import com.zeny.express.framework.shared.exception.ApplicationException;

public class ShardedMqClient<T> implements BeanNameAware {
	protected final static Logger LOGGER = LoggerFactory
			.getLogger(ShardedMqClient.class);

	private ShardIndexHelper getShardIndexGenerator = new ShardIndexHelper();
	private ShardIndexHelper putShardIndexGenerator = new ShardIndexHelper();
	private String beanName;

	public List<AbstractMqClient<T>> mqClientList;
	private long invalidRetryInterval = 30000;

	public List<AbstractMqClient<T>> getMqClientList() {
		return mqClientList;
	}

	public void setMqClientList(List<AbstractMqClient<T>> mqClientList) {
		this.mqClientList = mqClientList;
	}

	public long getErrorRetryInterval() {
		return invalidRetryInterval;
	}

	public void setErrorRetryInterval(long errorRetryInterval) {
		this.invalidRetryInterval = errorRetryInterval;
	}

	private AbstractMqClient<T> getShard(ShardIndexHelper shardIndexGenerator) {
		int index = shardIndexGenerator.getShardIndex(mqClientList.size());

		return mqClientList.get(index);
	}

	private void setInvalidShard(ShardIndexHelper shardIndexGenerator,
			AbstractMqClient<T> baseMqDAL) {
		int invalidIndex = mqClientList.indexOf(baseMqDAL);
		shardIndexGenerator.setInvalidShardIndex(invalidIndex);
	}

	public T get() {
		int curTryTime = 0;
		int errorTryTimes = 0;

		while (curTryTime < mqClientList.size()) {
			AbstractMqClient<T> baseMqDAL = getShard(getShardIndexGenerator);
			T data = null;

			try {
				data = baseMqDAL.get();
				if (data != null) {
					return data;
				} else {
					setInvalidShard(getShardIndexGenerator, baseMqDAL);
				}
			} catch (Exception e) {
				//TODO:沟通如何只识别MQException编码，需要指定
				errorTryTimes++;
				setInvalidShard(getShardIndexGenerator, baseMqDAL);
			}

			curTryTime++;
			LOGGER.error("{} Retry time {} of {}", beanName, curTryTime,
					mqClientList.size());
		}

		if (errorTryTimes >= mqClientList.size()) {
			//TODO:指定错误编码
			throw new ApplicationException(ApplicationException.UNKNOWN, null);
		}

		return null;
	}

	public void put(T data) {
		int curTryTime = 0;

		while (curTryTime < mqClientList.size()) {
			AbstractMqClient<T> baseMqDAL = getShard(putShardIndexGenerator);

			try {
				baseMqDAL.put(data);
				return;
			} catch (Exception e) {
				//TODO:沟通如何只识别MQException编码，需要指定
				setInvalidShard(putShardIndexGenerator, baseMqDAL);
			}

			curTryTime++;
			LOGGER.error("{} Retry time {} of {}", beanName, curTryTime,
					mqClientList.size());
		}

		//TODO:指定错误编码
		throw new ApplicationException(ApplicationException.UNKNOWN, null);
	}

	@Override
	public void setBeanName(String name) {
		this.beanName = name;
	}

	private class ShardIndexHelper {
		private int shardIndex = -1;
		private Map<Integer, Long> invalidMap = new HashMap<Integer, Long>();

		synchronized int getShardIndex(int shardSize) {
			int errorSize = invalidMap.size();

			while (true) {
				shardIndex = getNextShardIndex(shardSize, shardIndex);

				if ((errorSize == 0) || (errorSize == shardSize)
						|| !invalidMap.containsKey(shardIndex)) {
					// 1、全部正常，2、全部异常，3、当前节点无异常
					// 则依次返回
					break;
				}

				if (isRetryTime(shardIndex)) {
					invalidMap.remove(shardIndex);
					LOGGER.info("{} Retrying {}", beanName, shardIndex);
					break;
				}
			}

			return shardIndex;
		}

		synchronized void setInvalidShardIndex(int invalidShardIndex) {
			long errorTime = System.currentTimeMillis();
			invalidMap.put(invalidShardIndex, errorTime);

			LOGGER.info("{} invalid {}", beanName, shardIndex);
		}

		private int getNextShardIndex(int shardSize, int shardIndex) {
			if (shardIndex < shardSize - 1) {
				++shardIndex;
			} else {
				shardIndex = 0;
			}

			return shardIndex;
		}

		private boolean isRetryTime(int shardIndex) {
			long errorTime = invalidMap.get(shardIndex);
			long interval = System.currentTimeMillis() - errorTime;
			return interval > invalidRetryInterval;
		}
	}
}
